import json
from flask import Flask, render_template
from flask_socketio import SocketIO
from kafka import KafkaConsumer


app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)
thread = None
consumer = KafkaConsumer(
    auto_offset_reset='earliest',
    bootstrap_servers = "niit01:9092",
    group_id = "niit2",
    )
consumer.subscribe(["soutt6"])

def background_thread():
    girl = 0
    boy = 0
    # 0,1
    # 1,8
    # 2,1
    for msg in consumer:
        data_json = msg.value.decode('utf8')
        print('-----当前数据----'+ data_json )
        #  解析json数据
        # data_list = json.loads(data_json)
        obj = data_json.split(',')

        if obj[0]==1&obj[1]==0 in msg.value.decode('utf8'):
            girl = obj[2]
        if obj[0]==1& obj[1]==1in msg.value.decode('utf8'):
            boy = obj[2]
        else:
            continue
        result = str(girl) + ',' + str(boy)
        print(result)
        socketio.emit('test_message',{'data':result})


@socketio.on('test_connect')
def connect(message):
    print(message)
    global thread
    if thread is None:
        thread = socketio.start_background_task(target=background_thread)
    socketio.emit('connected', {'data': 'Connected'})


@app.route("/")
def handle_mes():
    return render_template("index.html")


if __name__ == '__main__':
    socketio.run(app,debug=True)